package com.shujia.streeam

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import scala.io.Source

object Demo12MakeData {
  def main(args: Array[String]): Unit = {

    /**
      * 模拟数据生成
      *
      * 数据打入kafka
      *
      */

    //建立连接

    val properties: Properties = new Properties()
    properties.setProperty("bootstrap.servers", "node2:9092,node3:9092,node4:9092") //kafka连接地址
    properties.put("key.serializer", classOf[StringSerializer])
    properties.put("value.serializer", classOf[StringSerializer])

    //创建生产者
    val producer = new KafkaProducer[String, String](properties)


    Source
      .fromFile("spark/data/dianxin_data")
      .getLines()
      .foreach(line => {

        println("正在发生数据：" + line)

        val record = new ProducerRecord[String, String]("dianxin", line)
        producer.send(record)


        Thread.sleep(100)
      })

    producer.close()

  }
}
